package march.firstmarch21.march03;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.*;

public class AcceptReducer extends Reducer<Text, Text,Text,IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        HashMap<String, Integer> map = new HashMap<>();
        for (Text value : values) {
            String phone = value.toString();
            int num = map.getOrDefault(phone, 0);
            num++;
            map.put(phone,num);
        }
        //Map对象化填充到List中
        List<Map.Entry<String,Integer>> list = new ArrayList<>(map.entrySet());

        //使用集合工具类Collections进行排序
        Collections.sort(list, new Comparator<Map.Entry<String, Integer>>() {
            @Override
            public int compare(Map.Entry<String, Integer> o1, Map.Entry<String, Integer> o2) {
                return o2.getValue() - o1.getValue();
            }
        });

        //将倒序好的集合进行输出，输出前三条数据
        int count = 0;
        for (Map.Entry<String, Integer> entry : list) {
            if(count < 3){
                count++;
                String phone = entry.getKey();
                int num = entry.getValue();
                context.write(new Text(key+","+phone),new IntWritable(num));
            }
        }
    }
}
